package cn.edu.laud.sany.dataimport.oracle;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.io.IOException;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import cn.edu.laud.sany.utils.Config;
import cn.edu.laud.sany.utils.MyLog;
import cn.edu.laud.sany.utils.ProcessManager;
import cn.edu.laud.sany.utils.TimeUtils;

public class OracleImport
{
	// config params
	public static int thread = 1;
	public static int port = 1521;
	public static String[] IP = new String[] {};
	public static String[] host = new String[] {};
	public static String database = "";
	public static String table = "gk";
	public static String csvFolder = "";
	public static String ignoreParaIdsFile = "";
	public static boolean errorDataStop = false;
	public static boolean ignoreSomeParaId = false;
	
	// used constants
	public static HashSet<Integer> ignoreParaIds = new HashSet<Integer>();
	public static String logFileFolder = "log";
	public static String logFileName = "mylog.txt";
	
	// running variables
	private static ThreadPoolExecutor executor;
	private static List<OracleRunner> runners;
	private static MyLog logger;
	public static Boolean halted = false;
	
	/**
	 * @param args
	 * @throws ClassNotFoundException
	 * @throws SQLException
	 */
	public static void main(String[] args) throws ClassNotFoundException, SQLException, IOException
	{
		initMyLog();
		doMain(args);
	}
	
	public static String getLogFilePath()
	{
		return (logFileFolder + "/" + logFileName);
	}
	
	protected static void initMyLog()
	{
		File logFolderFile = new File(logFileFolder);
		if (logFolderFile.exists() == false)
		{
			logFolderFile.mkdir();
		}
		logger = MyLog.getMyLog(getLogFilePath(), true);
	}
	
	protected static void doMain(String[] args) throws ClassNotFoundException, SQLException,
			IOException
	{
		if (args.length % 2 == 1 || args.length < 1)
		{
			System.err.println(Config.MySqlImportCommandHelpString);
			return;
		}
		
		try
		{
			parseParams(args);
		}
		catch (Exception e)
		{
			System.err.println(Config.MySqlImportCommandHelpString);
			return;
		}
		
		// test csv folder
		if (testCsvFoler() == false)
		{
			System.err.println("Can not find csv folder!");
			return;
		}
		if (ignoreSomeParaId && getIgnoreParaIds() == false)
		{
			System.err.println("Can not read ignoreParaIds correctly!");
			return;
		}
		
		try
		{
			Class.forName("oracle.jdbc.driver.OracleDriver");
			//java.sql.DriverManager.registerDriver(new oracle.jdbc.driver.OracleDriver());
		}
		catch (ClassNotFoundException e)
		{
			e.printStackTrace();
			return;
		}
		catch (Exception e)
		{
			e.printStackTrace();
			return;
		}
		
		try
		{
			ProcessManager.InitProcessManager(logger, csvFolder, logFileFolder, false);
			Thread.sleep(3000);
		}
		catch (Exception e)
		{
			System.err.println("Init ProcessManager failed: " + e.getMessage());
			return;
		}
		
		long starttime = System.nanoTime();
		
		logger.info("Create threads to import data to LaUD...");
		
		BlockingQueue<Runnable> queue = new LinkedBlockingQueue<Runnable>();
		executor = new ThreadPoolExecutor(thread, thread, 100, TimeUnit.DAYS, queue);
		runners = new ArrayList<OracleRunner>();
		
		for (int i = 0; i < thread; i++)
		{
			OracleRunner runner = new OracleRunner(i, database);
			runners.add(runner);
			executor.execute(runner);
		}
		executor.shutdown();
		try
		{
			executor.awaitTermination(100, TimeUnit.DAYS);
		}
		catch (InterruptedException e1)
		{
			logger.error("Main thread interrupted while waiting " + e1.getMessage());
		}
		
		ProcessManager.applicationExit = true;
		long endTime = System.nanoTime();
		logger.info("Import program's total run time:"
				+ TimeUtils.nanoSecondToString((endTime - starttime)));
	}
	
	protected static void parseParams(String[] args)
	{
		for (int i = 0; i < args.length; i += 2)
		{
			if (args[i].equals("--thread"))
			{
				thread = Integer.valueOf(args[i + 1]);
			}
			else if (args[i].equals("--port"))
			{
				port = Integer.valueOf(args[i + 1]);
			}
			else if (args[i].equals("--ip"))
			{
				IP = args[i + 1].split(",");
				host = new String[IP.length];
				for (int k = 0; k < IP.length; k++)
				{
					host[k] = String.format(Config.OracleJDBCFormat, IP[k], port);
				}
			}
			else if (args[i].equals("--database"))
			{
				database = args[i + 1];
			}
			else if (args[i].equals("--csvpath"))
			{
				csvFolder = args[i + 1];
			}
			else if (args[i].equals("--ignorefile"))
			{
				ignoreSomeParaId = true;
				ignoreParaIdsFile = args[i + 1];
			}
			else if (args[i].equals("--errorStop"))
			{
				if (args[i + 1].equals("1"))
				{
					errorDataStop = true;
				}
			}
			else if (args[i].equals("--table"))
			{
				table = args[i + 1];
			}
			else if (args[i].equals("--batch"))
			{
				OracleRunner.JDBC_BATCH_SIZE = Integer.valueOf(args[i + 1]);
			}
		}
		logger.info("Parameters are:");
		logger.info("thread = " + thread);
		logger.info("port = " + port);
		String ipsString = "";
		for (String ip : IP)
		{
			ipsString += ip + "; ";
		}
		logger.info("ip = " + ipsString);
		logger.info("keyspace = " + database);
		logger.info("csv path = " + csvFolder);
		logger.info("ignore file = " + ignoreParaIdsFile);
		logger.info("errorData Stop = " + errorDataStop);
		logger.info("table = " + table);
		logger.info("JDBC_BATCH_SIZE = " + OracleRunner.JDBC_BATCH_SIZE);
	}
	
	protected static boolean testCsvFoler()
	{
		try
		{
			File file = new File(csvFolder);
			if (file.exists())
			{
				return true;
			}
		}
		catch (Exception e)
		{
			e.printStackTrace();
		}
		return false;
	}
	
	protected static boolean getIgnoreParaIds()
	{
		ignoreParaIds = new HashSet<Integer>();
		boolean flag = true;
		File file = new File(ignoreParaIdsFile);
		BufferedReader bReader = null;
		String paraIdString = null;
		try
		{
			bReader = new BufferedReader(new FileReader(file));
			while ((paraIdString = bReader.readLine()) != null)
			{
				paraIdString = paraIdString.trim();
				if (paraIdString.equals(""))
				{
					continue;
				}
				int paraId = Integer.valueOf(paraIdString);
				ignoreParaIds.add(paraId);
			}
			
		}
		catch (IOException e1)
		{
			flag = false;
			logger.error("Can not open ignoreParaIds file!");
		}
		catch (NumberFormatException e2)
		{
			flag = false;
			logger.error("IgnoreParaIds file contains invalid integer!");
		}
		finally
		{
			if (flag == false)
			{
				logger.info("Get ignoreParaIds failed!");
			}
			try
			{
				if (bReader != null)
				{
					bReader.close();
				}
			}
			catch (IOException e)
			{
				logger.info(e.getMessage());
			}
		}
		return flag;
	}
	
	public static void stopProgram()
	{
		halted = true;
		logger.info("Try to stop import program...");
		for (OracleRunner runner : runners)
		{
			executor.remove(runner);
			runner.interrupt();
		}
		System.exit(1);
	}
	
}
